/*
 * Copyright 2022 LMAX Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.sgy.disruptor;

import java.util.concurrent.atomic.AtomicInteger;

import static cn.sgy.disruptor.RewindAction.REWIND;
import static java.lang.Math.min;


/**
 * 带有批处理效果的执行消费Runnable
 * 何为批处理？
 * Convenience class for handling the batching semantics of consuming entries from a {@link RingBuffer}
 * and delegating the available events to an {@link EventHandler}.
 *
 * @param <T> event implementation storing the data for sharing during exchange or parallel coordination of an event.
 */
public final class BatchEventProcessor<T>
        implements EventProcessor {
    private static final int IDLE = 0;
    private static final int HALTED = IDLE + 1;
    private static final int RUNNING = HALTED + 1;
    /**
     * 运行状态
     */
    private final AtomicInteger running = new AtomicInteger(IDLE);
    /**
     * 消费异常处理器
     */
    private ExceptionHandler<? super T> exceptionHandler;
    /**
     * 消费者要消费数据，总得从队列中拿吧，这个实际上就是RingBuffer
     * 为什么是DataProvider类型？因为我只希望消费者能调用获取数据的get方法即可，其他方法我不想让消费者知道
     */
    private final DataProvider<T> dataProvider;
    /**
     * 消费者序号要根生产者序号进行对比，然后进行一些处理，这个就是干这个的
     */
    private final SequenceBarrier sequenceBarrier;
    /**
     * 用于消费队列数据的接口
     */
    private final EventHandlerBase<? super T> eventHandler;
    private final int batchLimitOffset;
    /**
     * 消费者序号
     */
    private final Sequence sequence = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
    private final RewindHandler rewindHandler;
    private int retriesAttempted = 0;

    BatchEventProcessor(
            final DataProvider<T> dataProvider,
            final SequenceBarrier sequenceBarrier,
            final EventHandlerBase<? super T> eventHandler,
            final int maxBatchSize,
            final BatchRewindStrategy batchRewindStrategy
    ) {
        // 初始化没啥好讲的
        this.dataProvider = dataProvider;
        this.sequenceBarrier = sequenceBarrier;
        this.eventHandler = eventHandler;

        if (maxBatchSize < 1) {
            throw new IllegalArgumentException("maxBatchSize must be greater than 0");
        }
        this.batchLimitOffset = maxBatchSize - 1;
        this.rewindHandler = eventHandler instanceof RewindableEventHandler
                ? new TryRewindHandler(batchRewindStrategy)
                : new NoRewindHandler();
    }

    @Override
    public Sequence getSequence() {
        return sequence;
    }

    @Override
    public void halt() {
        running.set(HALTED);
        sequenceBarrier.alert();
    }

    @Override
    public boolean isRunning() {
        return running.get() != IDLE;
    }

    /**
     * Set a new {@link ExceptionHandler} for handling exceptions propagated out of the {@link BatchEventProcessor}.
     *
     * @param exceptionHandler to replace the existing exceptionHandler.
     */
    public void setExceptionHandler(final ExceptionHandler<? super T> exceptionHandler) {
        if (null == exceptionHandler) {
            throw new NullPointerException();
        }

        this.exceptionHandler = exceptionHandler;
    }

    /**
     * It is ok to have another thread rerun this method after a halt().
     *
     * @throws IllegalStateException if this object instance is already running in a thread
     */
    @Override
    public void run() {
        int witnessValue = running.compareAndExchange(IDLE, RUNNING);
        // Successful CAS
        if (witnessValue == IDLE) {
            sequenceBarrier.clearAlert();

            notifyStart();
            try {
                if (running.get() == RUNNING) {
                    // 开始消费队列中的数据
                    processEvents();
                }
            } finally {
                notifyShutdown();
                running.set(IDLE);
            }
        } else {
            if (witnessValue == RUNNING) {
                throw new IllegalStateException("Thread is already running");
            } else {
                earlyExit();
            }
        }
    }

    private void processEvents() {
        // 从队列中获取出来的元素
        T event = null;
        // 开始要消费的序号
        long nextSequence = sequence.get() + 1L;
        // 死循环消费
        while (true) {
            // 开始要消费的序号
            final long startOfBatchSequence = nextSequence;
            try {
                try {
                    // 开始消费之前，和生产者序号对比一下，看看是否要去阻塞，消费者序号大于生产者序号就不能消费
                    // 返回的是可以消费的最大进度，就是生产者的进度
                    final long availableSequence = sequenceBarrier.waitFor(nextSequence);
                    // 结束消费的序号
                    final long endOfBatchSequence = min(nextSequence + batchLimitOffset, availableSequence);
                    // 如果开始消费的序号小于等于结束消费的序号，比如2-10，这里体现了批处理
                    if (nextSequence <= endOfBatchSequence) {
                        // 调用EventHandler进行批量消费
                        eventHandler.onBatchStart(endOfBatchSequence - nextSequence + 1, availableSequence - nextSequence + 1);
                    }
                    // 如果开始消费的序号小于等于结束消费的序号，比如2-10，这里体现了批处理
                    while (nextSequence <= endOfBatchSequence) {
                        // 从队列中获取元素
                        event = dataProvider.get(nextSequence);
                        // 消费
                        eventHandler.onEvent(event, nextSequence, nextSequence == endOfBatchSequence);
                        // 推进下一个消费的序号
                        nextSequence++;
                    }

                    retriesAttempted = 0;
                    // 消费者序号更新
                    sequence.set(endOfBatchSequence);
                } catch (final RewindableException e) {
                    nextSequence = rewindHandler.attemptRewindGetNextSequence(e, startOfBatchSequence);
                }
            } catch (final TimeoutException e) {
                notifyTimeout(sequence.get());
            } catch (final AlertException ex) {
                if (running.get() != RUNNING) {
                    break;
                }
            } catch (final Throwable ex) {
                handleEventException(ex, nextSequence, event);
                sequence.set(nextSequence);
                nextSequence++;
            }
        }
    }

    private void earlyExit() {
        notifyStart();
        notifyShutdown();
    }

    private void notifyTimeout(final long availableSequence) {
        try {
            eventHandler.onTimeout(availableSequence);
        } catch (Throwable e) {
            handleEventException(e, availableSequence, null);
        }
    }

    /**
     * Notifies the EventHandler when this processor is starting up.
     */
    private void notifyStart() {
        try {
            eventHandler.onStart();
        } catch (final Throwable ex) {
            handleOnStartException(ex);
        }
    }

    /**
     * Notifies the EventHandler immediately prior to this processor shutting down.
     */
    private void notifyShutdown() {
        try {
            eventHandler.onShutdown();
        } catch (final Throwable ex) {
            handleOnShutdownException(ex);
        }
    }

    /**
     * Delegate to {@link ExceptionHandler#handleEventException(Throwable, long, Object)} on the delegate or
     * the default {@link ExceptionHandler} if one has not been configured.
     */
    private void handleEventException(final Throwable ex, final long sequence, final T event) {
        getExceptionHandler().handleEventException(ex, sequence, event);
    }

    /**
     * Delegate to {@link ExceptionHandler#handleOnStartException(Throwable)} on the delegate or
     * the default {@link ExceptionHandler} if one has not been configured.
     */
    private void handleOnStartException(final Throwable ex) {
        getExceptionHandler().handleOnStartException(ex);
    }

    /**
     * Delegate to {@link ExceptionHandler#handleOnShutdownException(Throwable)} on the delegate or
     * the default {@link ExceptionHandler} if one has not been configured.
     */
    private void handleOnShutdownException(final Throwable ex) {
        getExceptionHandler().handleOnShutdownException(ex);
    }

    private ExceptionHandler<? super T> getExceptionHandler() {
        ExceptionHandler<? super T> handler = exceptionHandler;
        return handler == null ? ExceptionHandlers.defaultHandler() : handler;
    }

    private class TryRewindHandler implements RewindHandler {
        private final BatchRewindStrategy batchRewindStrategy;

        TryRewindHandler(final BatchRewindStrategy batchRewindStrategy) {
            this.batchRewindStrategy = batchRewindStrategy;
        }

        @Override
        public long attemptRewindGetNextSequence(final RewindableException e, final long startOfBatchSequence) throws RewindableException {
            if (batchRewindStrategy.handleRewindException(e, ++retriesAttempted) == REWIND) {
                return startOfBatchSequence;
            } else {
                retriesAttempted = 0;
                throw e;
            }
        }
    }

    private static class NoRewindHandler implements RewindHandler {
        @Override
        public long attemptRewindGetNextSequence(final RewindableException e, final long startOfBatchSequence) {
            throw new UnsupportedOperationException("Rewindable Exception thrown from a non-rewindable event handler", e);
        }
    }
}